/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.iterative;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.DatasetFactory;
import cgl.granules.dataset.FileAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceException;
import cgl.granules.operation.ProgressTracker;
import cgl.granules.operation.InstanceDeployer;

/**
 * The deployer for the example demonstrating iterative semantics using
 * Granules. The example demonstrates the orchestration of iterative execution
 * of these stages until a termination condition has been reached.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class IterativeDeployer extends InstanceDeployer {
    private ProgressTracker progressTracker;


    /**
     * Initializes the deployer.
     * 
     * @param streamingProperties
     *            The properties for streaming
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws IOException
     *             If there are IO problems
     * @throws MarshallingException
     *             If there are marshalling erros
     * @throws DeploymentException
     *             If there are deployment related problems
     */
    public IterativeDeployer(Properties streamingProperties)
        throws CommunicationsException, IOException, MarshallingException,
        DeploymentException {
        initialize(streamingProperties);
    }


    /**
     * Initialize a set of <code>StageTwo</code>s. Each of the
     * <code>StageTwo</code> instances have inputs from <code>StageOne</code>
     * and send their outputs to the <code>StageThree</code> instance.
     * 
     * @param stageOne
     *            The <code>StageOne</code> instance
     * @param stageThree
     *            The <code>StageThree</code> instance
     * @param numOfStageTwos
     *            The number of <code>StageTwo</code>s that need to be
     *            initialized.
     * @param filenameBase
     *            The directory in which the files are stored.
     * @param numOfFiles
     *            The total number of files that need to be processed
     * @return The initialized set of code>StageTwo[]</code> instances.
     * @throws DatasetException
     *             If there are problems initializing/using the datasets
     * @throws MapReduceException
     *             If there are problems related to Map/Reduce roles within the
     *             application.
     */
    public StageTwo[] initializeStageTwos(StageOne stageOne,
                                          StageThree stageThree,
                                          int numOfStageTwos,
                                          String filenameBase, int numOfFiles)
        throws DatasetException, MapReduceException {

        stageThree.addReduce(stageOne);
        stageThree.setAsExecuteWhenDataAvailable();

        StageTwo[] maps = new StageTwo[numOfStageTwos];
        String fileDatasetIdentifier = "FileDataset-ID-IterativeStageTwo";
        for (int i = 0; i < numOfStageTwos; i++) {
            maps[i] = new StageTwo();
            DatasetFactory datasetFactory = DatasetFactory.getInstance();
            FileAccess fileAccess =
                datasetFactory.createFileDataset(fileDatasetIdentifier,
                                                 "Iterative example");

            for (int j = 0; j < numOfFiles; j++) {
                if (j % numOfStageTwos == i) {
                    String filename = filenameBase + j + ".txt";
                    // System.out.println(filename + "-> assigned to MAP [" + i
                    // + "]");
                    fileAccess.addFile(filename, "rw");
                }/* the modulo-check */
            }

            DatasetCollection datasetCollection =
                maps[i].getDatasetCollection();
            datasetCollection.addDataset(fileAccess);

            if (datasetCollection.hasDataset(fileDatasetIdentifier)) {
                System.out.println("Added dataset [" + fileDatasetIdentifier
                    + "] to map (" + i + ")");
            }

            stageOne.addReduce(maps[i]);
            maps[i].addReduce(stageThree);
            maps[i].setAsExecuteWhenDataAvailable();
        }

        return maps;
    }


    /**
     * Prepares and deploys an iterative pipeline comprising one
     * <code>StageOne</code> and <code>StageThree</code> instances, and the
     * specified number of <code>StageTwo</code>s.
     * 
     * @param numOfStageTwos
     *            The number of <code>StageTwo</code>s that need to be
     *            initialized.
     * @param filenameBase
     *            The directory in which the files are stored.
     * @param numOfFiles
     *            The total number of files that need to be processed
     * @throws DatasetException
     *             If there are problems initializing the datasets
     * @throws MapReduceException
     *             If there is a MapReduceExcecution because of problems when
     *             setting up the roles.
     * @throws IOException
     *             If there are IO problems.
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws DeploymentException
     *             If there are deployment related problems
     * @throws MarshallingException
     *             If there are marshalling problems.
     * @throws InterruptedException
     *             If there are problems sleeping
     */
    public void prepareInstancesAndDeploy(int numOfStageTwos,
                                          String filenameBase, int numOfFiles)
        throws MapReduceException, CommunicationsException,
        DeploymentException, MarshallingException, DatasetException,
        InterruptedException {

        StageOne stageOne = new StageOne();
        StageThree stageThree = new StageThree();

        StageTwo[] stageTwos =
            initializeStageTwos(stageOne, stageThree, numOfStageTwos,
                                filenameBase, numOfFiles);

        StageOne[] stageOnes = { stageOne };
        resetTrackers();

        progressTracker = deployOperations(stageOnes);
        String jobId = progressTracker.getIdentifier();

        deployOperations(stageTwos, jobId);

        StageThree[] stageThrees = { stageThree };
        deployOperations(stageThrees, jobId);

    }


    /**
     * @param args
     */
    public static void main(String[] args) {
        String module = "IterativeDeployer.main()->";
        if (args.length < 2) {
            System.out.println("Usage: java cgl.granules.samples.periodic"
                + "IterativeDeployer <hostname> <port>");
            System.exit(0);
        }
        String hostname = args[0];
        String portnum = args[1];
        int numOfStageTwos = 2;
        int numOfFiles = 2;

        String filenameBase = "C:/Granules/Datasets/Files/file";

        Properties streamingProperties = new Properties();
        streamingProperties.put("hostname", hostname);
        streamingProperties.put("portnum", portnum);

        String helpInfo =
            "\n\nThis is the word count test program that uses Granules."
                + "The current list of commands include: \n"
                + "submit <filename_base> <num_of_maps> <num_of_resources> \n"
                + "status       : Lists the status of tasks that comprise Job"
                + "h            : List the set of commands \n\n";
        try {
            IterativeDeployer deployer =
                new IterativeDeployer(streamingProperties);
            StringTokenizer tok;

            InputStreamReader isr = new InputStreamReader(System.in);
            BufferedReader commandLine = new BufferedReader(isr);

            System.out.println(helpInfo);

            while (true) {
                String s = commandLine.readLine();
                if (s == null) {
                    System.out.println(module + "String is null!!!");
                    break;
                }

                if (s.startsWith("submit")) {
                    System.out.println(module
                        + "Processing the [submit] command");

                    tok = new StringTokenizer(s, " ");
                    int numOfTokens = tok.countTokens();
                    tok.nextToken();/* skip the command */

                    if (numOfTokens > 1) {
                        filenameBase = tok.nextToken();
                    }

                    if (numOfTokens > 2) {
                        numOfStageTwos = Integer.parseInt(tok.nextToken());
                    }

                    if (numOfTokens > 3) {
                        numOfFiles = Integer.parseInt(tok.nextToken());
                    }

                    String params =
                        "\nFilename base      : " + filenameBase
                            + "\nNumber of StageTwos: " + numOfStageTwos
                            + "\nNumber of Files    :" + numOfFiles;
                    System.out.println(params);

                    deployer.prepareInstancesAndDeploy(numOfStageTwos,
                                                       filenameBase, numOfFiles);

                    continue;
                }

                if (s.startsWith("status")) {
                    deployer.getStatus();
                    continue;
                }
            }

        } catch (Exception e) {
            System.out.println(e.getMessage());
            e.printStackTrace();
        }
    }
}
